package com.zhangtai.mq;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.UUID;

/**
 * @Author: ming
 * @Date: 2020/5/31 0031 下午 1:49
 */
@Component
@Slf4j
public class MessageSender implements RabbitTemplate.ConfirmCallback , RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RedisTemplate redisTemplate;


    @PostConstruct
    public void init(){

        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }


    /*
     * 如果消息没有到达交换机,则该方法中isSendSuccess = false, error为错误信息;
     * 如果消息正确到达交换机,则该方法中isSendSuccess = true;
     */

    @Override
    public void confirm(CorrelationData correlationData, boolean isSendSuccess, String error) {
        if (isSendSuccess){
            log.info("发送消息正确到到达交换机，发送成功,id是：{}",correlationData);
            redisTemplate.delete(correlationData.getId());
        }

        if(!isSendSuccess){
            log.info("消息未到达交换机，发送失败,id是{}，原因是：{}",correlationData.getId(),error);
            HashMap<String,String> map = (HashMap<String, String>) redisTemplate.boundValueOps(correlationData.getId()).get();
            String exchange = map.get("exchange");
            String routingKey = map.get("routingKey");
            String message = map.get("message");
            //投递失败，重新发送
            rabbitTemplate.convertAndSend(exchange,routingKey, JSON.toJSONString(message));
        }

    }

    /*
     * 消息从交换机成功到达队列，则returnedMessage方法不会执行;
     * 消息从交换机未能成功到达队列，则returnedMessage方法会执行;
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

        log.info("投递失败的消息是：{}",new String(message.getBody(), StandardCharsets.UTF_8));
        rabbitTemplate.convertAndSend(exchange,routingKey, JSON.toJSONString(message));
    }

    /**
     * 发送消息
     * @param exchange
     * @param routingKey
     * @param message
     */
    public void send(String exchange,String routingKey,String message){
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        HashMap<String, String> hashMap = new HashMap<>();
        hashMap.put("exchange",exchange);
        hashMap.put("routingKey",routingKey);
        hashMap.put("message",message);
        redisTemplate.boundValueOps(correlationData.getId()).set(hashMap);
        rabbitTemplate.convertAndSend(exchange,routingKey,message,correlationData);
    }


}
